package com.foreveross.proxyip.core.zk;

import java.io.IOException;
import java.lang.reflect.Method;
import java.net.InetAddress;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import net.sf.json.JSONObject;

import org.apache.commons.lang3.StringUtils;
import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.SessionExpiredException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

import com.foreveross.proxyip.core.BaseCode;
import com.foreveross.proxyip.core.redis.PiRedisExecption;
import com.foreveross.proxyip.core.redis.RedisExecutor;
import com.foreveross.proxyip.core.redis.RedisLock;
import com.foreveross.proxyip.core.zk.task.NormalTask;
import com.foreveross.proxyip.core.zk.task.ZkTask;
import com.foreveross.proxyip.core.zk.task.ZkTaskFactory;
import com.foreveross.proxyip.infra.utils.ThreadPoolUtil;
import com.foreveross.proxyip.util.SerializeUtil;

/**
 * 
 * @author lyq
 * 
 */
public class ZkAppCenter implements Watcher {

	boolean sessionTimeout = false;
	//private static LatchControl latchControl = new LatchControl(30000L);

	private static final String CREATE_TASK_LOCK = "create_task_lock";

	private static volatile ZkAppCenter instance = null;

	// private static ZkAppInfo info = null;
	protected static final Logger log = Logger.getLogger(ZkAppCenter.class);

	public static final String appGroup = "/proxyipAPP";

	public static final String normalTask = "/normalTask";

	private static final String REG_LOCK_NAME = appGroup + "lock";
	
	private static final String IPPROXY_APP_SIZE = "ipproxy_app_size";
	
	public String tasksPath = "/proxyip_tasks";
	public String leaderPath = "/leader";

	String lead = null;

	public Set<String> taskSet = new HashSet<String>();

	// 每个任务在execute中是加锁的，不能并行执行某一任务，这个键值谁不会被覆盖
	// markfb
	 private final Map<String, CountDownLatch> taskLatchMap = new HashMap<String, CountDownLatch>();

	// ZkLock regLock = null,taskLock = null;

	Map<String, ZkLock> locks = new HashMap<String, ZkLock>();

	private ZkAppCenter() {
		init();
	}

	private ZkAppCenter(String[] tasks) {
		for (String task : tasks) {
			taskSet.add(task);
		}
	}

	public void init() {
		// String [] tasks = {PublicPool.POOL_CODE+"_DoloadTask"
		// ,PublicPool.POOL_CODE+"_CheckInTask"
		// ,PublicPool.POOL_CODE+"_CheckOutTask"
		// ,ChannelPool.DEFAULT_CODE
		// ,DbPool.POOL_CODE+"_CheckInTask"
		// ,DbPool.POOL_CODE+"_CheckOutTask"
		// ,DbPool.POOL_CODE+"_DoloadTask"
		// ,"task_TestTask","task1_TestTask","task2_TestTask","task3_TestTask","task4_TestTask","task5_TestTask","task6_TestTask"};
		// this.tasks = tasks;
	}

	private ZooKeeper zk = null;

	public String appName = null;

	public static ZkAppCenter getInstance() {
		if (instance == null) {
			
			synchronized (ZkAppCenter.class) {
				if (instance == null) {
					instance = new ZkAppCenter();
				}
			}
		}
		
		return instance;
	}

	public static ZkAppCenter getInstance(String[] appTask) {
		if (instance == null) {

			synchronized (ZkAppCenter.class) {
				if (instance == null) {
					instance = new ZkAppCenter(appTask);
				}
			}
		}

		return instance;
	}

	// public static ZkAppInfo getInstance(){
	// if(info == null){
	// info = new ZkAppInfo();
	// }
	// return info;
	// }
	public ZooKeeper getZK() throws Exception {
		//latchControl.waitLatch();
		return newZk();
	}

	private ZooKeeper newZk() throws PiRedisExecption, IOException {
		Integer SESSION_TIMEOUT = null;
		String hosts = null;
		
		if (zk == null) {
			synchronized (ZkAppCenter.class) {
				if (zk == null) {
					try {
						SESSION_TIMEOUT = Integer.parseInt(BaseCode.getValue("ZK_SESSION_TIMEOUT"));
						hosts = BaseCode.getValue("ZK_HOST");
						zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
						TimeUnit.SECONDS.sleep(1);
					} catch (Exception e) {
						log.error("zk创建异常：", e);
					}
				}
			}
		}
		
		return zk;
	}

	static Boolean isRegistration = false;// 注册中
	/**
	 * 注册每个应用是单线程进行的，如果有两个线程并行执行注册操作，那么只有一个执行线程有效。
	 * 
	 * @return
	 * @throws Exception
	 */
	public CountDownLatch regLatch = null;

	public String registerApp() throws Exception {
		try {
			synchronized (isRegistration) {
				if (isRegistration) {
					log.info("正在注册中.............................................");
					if (regLatch == null) {
						regLatch = new CountDownLatch(1);
					}
					regLatch.await(10000, TimeUnit.MILLISECONDS);// 等待注册完成
					return appName;
				}
				
				this.locks.clear();
				taskLatchMap.clear();
				this.zk = null;
				isRegistration = true;
			}
			Thread.sleep(4000); // 等待一下，看看有没有并发线程，让并发注册操作'regLatch.wait(10000);'
								// ；同时如果该节点是主节点，可以让主节点有时间删除所有创建路径
			log.info("=================register start0========================");
			newZk();
			if (zk.exists(appGroup, false) == null) {
				zk.create(appGroup, appGroup.getBytes(), Ids.OPEN_ACL_UNSAFE,
						CreateMode.PERSISTENT);
			}
			log.info("------------------------------register 1----------------------------------");
			ZkLock regLock = locks.get(REG_LOCK_NAME);
			if (regLock == null) {
				log.info("------------------------------register 2----------------------------------");
				RedisLock lock = new RedisLock("registerApp_Lock");// 需要加锁，不能并行。
				// regLock = new ZkLock(REG_LOCK_NAME,zk) {//创建应用，和任务链的构建必须是有序的。
				// @Override
				// public Object doAction() {
				try {
					lock.lock(1000);
					log.info("------------------------------register 3----------------------------------");
					if (zk.exists(normalTask, false) == null) {// 常规任务发布地址
						zk.create(normalTask, normalTask.getBytes(),
								Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
					}
					zk.getData(normalTask, true, null);

					if (zk.exists(tasksPath, false) == null) {
						zk.create(tasksPath, tasksPath.getBytes(),
								Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
					}
					String addr = InetAddress.getLocalHost().getHostAddress();
					if (appName == null || zk.exists(appName, false) == null) {
						appName = zk.create(appGroup + "/" + addr + "-",
								new byte[0], Ids.OPEN_ACL_UNSAFE,
								CreateMode.EPHEMERAL_SEQUENTIAL);
					}
					// 如果是主机，那么将主机创建领导路径；
					// 所有节点都监听主机创建的领导路径，因为主机创建了所有任务路径；
					// 主机挂掉的时候，所有应用将重新抢着争做主机；
					// leaderPath = getTaskPath(zk,"register");//

					if (zk.exists(leaderPath, false) == null) {
						zk.create(leaderPath, appName.getBytes(),
								Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
						new RedisExecutor<Boolean>() {
							@Override
							public Boolean run() throws Exception {
								this.jedis.flushAll();
								return null;
							}
						}.exec();
					}

					lead = new String(zk.getData(leaderPath, true, null));
					log.info("leader :" + lead);
					for (String task : taskSet) {
						String taskPath = this.getTaskPath(zk, task);
						zk.getData(taskPath, true, null);
					}
					log.debug("---------------------------registerApp end-------------------------------");

					// listenChain();
					// listenLeader();//监听leader节点
					return null;
				} catch (Exception e) {
					throw new RuntimeException(e);
				} finally {
					lock.unlock();
				}// 每个应用注册进来必须重新构造一遍选举监听链
					// }
				// };
				// locks.put(REG_LOCK_NAME, regLock);
			}
			// regLock.execute(100000L);

			return appName;
		} finally {
			isRegistration = false;
			if (regLatch != null) {
				regLatch.countDown();
			}
		}
	}

	public void listenTask(String taskPaths) throws Exception {
		String[] tasks = StringUtils.split(taskPaths, ",");
		for (String taskPath : tasks) {
			getZK().getData(taskPath, true, null);// 监听该任务
		}
	}


	/**
	 * 发布一条任务路径，如果存在不创建
	 */
	public String getTaskPath(ZooKeeper zk, String task) throws Exception {
		String taskPath = this.tasksPath + "/" + task;
		if (zk.exists(taskPath, false) == null) {
			zk.create(taskPath, new byte[] {}, Ids.OPEN_ACL_UNSAFE,
					CreateMode.EPHEMERAL);
		}
		return taskPath;
	}


	@SuppressWarnings("unchecked")
	private void doTask(String taskParams) throws Exception {
		Map<String, Object> tparams = JSONObject.fromObject(taskParams);
		String[] params = StringUtils.split(tparams.get("taskParams").toString(), "_");
		String taskClass = params[1].toString();
		ZkTask taskBean = ZkTaskFactory.getInstance().getTask(taskClass);
		int result = taskBean.execute(tparams);
		log.debug(taskParams + " :taskParams:" + result);
		
		if("DoloadTask".equals(taskClass)) {
			// 这里对于任务的完成不是很准确，这里只能说明当前这个应用
			// 下载ip完成，其它的应用不一定完成了，但相对固定休眠时间要好点
			CountDownLatch  cdl = taskLatchMap.get(taskParams);
			if (cdl != null) {
				cdl.countDown();
			}
		}
	}

	/**
	 * 等待任务全部执行完,这里所有应用执行某一任务是加锁的； 要么超过五秒，要么分布式任务全部执行完。锁解开
	 * 
	 * @param execTaskParams
	 * @return
	 * @return
	 * @throws Exception
	 */
	public void executeWait(String params, String taskParams) throws Exception {
		// final String taskParams = taskCode + "_" + execClassName;
		try {
			// markfb
			CountDownLatch latch = new CountDownLatch(1);
			taskLatchMap.put(params, latch);
			 
			String taskPath = this.getTaskPath(this.getZK(), taskParams);
			this.getZK().setData(taskPath, params.getBytes(), -1);

			// markfb
			if("DoloadTask".equals(taskParams.split("_")[1])) {
				latch.await(60L, TimeUnit.SECONDS);
			}
			
			taskLatchMap.remove(params);
		} catch (Exception e) {
			if (e instanceof SessionExpiredException) {
				this.reRegister();
			}
		}
	}

	/**
	 * 不用等待，直接执行。
	 * 
	 * @param task
	 * @throws Exception
	 */
	public void execute(String params, String taskParams) throws Exception {
		// final String taskParams = taskCode + "_" + execClassName;
		try {
			// markfb
			/*
			 * CountDownLatch latch = new CountDownLatch(1);
			 * taskLatchMap.put(taskParams, latch);
			 */
			String taskPath = this.getTaskPath(this.getZK(), taskParams);
			this.getZK().setData(taskPath, params.getBytes(), -1);

			// markfb
			// latch.await(500L, TimeUnit.SECONDS);
			// taskLatchMap.remove(taskParams);
		} catch (Exception e) {
			if (e instanceof SessionExpiredException) {
				this.reRegister();
			}
		}
	}

	/**
	 * 任务链监听构造，编号大的app监听标号小一位的app，最小标号那个app不监听任何app，最大编号那个没有任何app监听
	 * 
	 * @throws KeeperException
	 * @throws InterruptedException
	 * @throws Exception
	 */
	@SuppressWarnings("unused")
	@Deprecated
	private void listenChain() throws KeeperException, InterruptedException,
			Exception {
		List<String> apps = this.getZK().getChildren(appGroup, false);
		Collections.sort(apps);
		for (int i = apps.size() - 1; i > 0; i--) {
			if (appName.contains(apps.get(i))) {
				String listen = apps.get(i - 1);
				this.getZK().getData(appGroup + "/" + listen, true, null);// 监听比自己小的一位
				break;
			}
		}
	}

	public Integer getAppsSize() throws Exception {
		try {
			List<String> list = getApps();
			Integer appsSize = list == null ? 0 : list.size();
			
			if (appsSize == 0) {
				log.error("集群失败，没有'" + appGroup + "'该应用群组里。");
				throw new PiZkException("应用注册丢失，重新注册");
			}
			
			return appsSize;
		} catch (Exception e) {
			this.reRegister();// 虽然解决了故障，已经抛出异常，这是一个全新的开始。
			throw e;
		}
	}

	public List<String> getApps() throws Exception {
		try {
			return new RedisExecutor<List<String>>() {
				@Override
				public List<String> run() throws Exception {
					@SuppressWarnings("unchecked")
					List<String> apps = (List<String>) SerializeUtil.unserialize(jedis.get(IPPROXY_APP_SIZE.getBytes()));
					
					if (apps == null || apps.isEmpty()) {
						if (StringUtils.isNotBlank(appName) && getZK().exists(appName, false) == null) {
							log.error(appName + "id noy exist...");
							reRegister();
						}
						
						apps = getZK().getChildren(appGroup, false);
						jedis.setex(IPPROXY_APP_SIZE.getBytes(), 10, SerializeUtil.serialize(apps));
					}
					
					return apps;
				}
			}.exec();
		} catch (Exception e) {
			this.reRegister();// 虽然解决了故障，已经抛出异常，这是一个全新的开始。
			throw e;
		}
	}

	public void doNormalTask(String task) throws Exception {
		this.getZK().setData(normalTask, task.getBytes(), -1);
	}

	private void doInnerNormalTask(String task) throws Exception {
		String[] params = StringUtils.split(task, "!");
		NormalTask taskExecutor = new NormalTask();
		Method method = null;
		if (params.length < 2) {
			method = NormalTask.class.getDeclaredMethod(params[0]);
		} else {
			method = NormalTask.class.getDeclaredMethod(params[0], String.class);
		}
		method.invoke(taskExecutor, params.length < 2 ? new Object[] {} : params[1]);
	}

	@Override
	public void process(WatchedEvent event) {
		try {
			final String eventPath = event.getPath();
			
			if (StringUtils.isBlank(eventPath)) {
				return;
			}
			
			if (eventPath != null && eventPath.contains("lock")) {
				if (eventPath.contains(REG_LOCK_NAME)) {
					locks.get(REG_LOCK_NAME).latch.countDown();
				} else if (eventPath.contains(CREATE_TASK_LOCK)) {
					locks.get(CREATE_TASK_LOCK).latch.countDown();
				} else {

					String lockName = StringUtils.split(eventPath, "/")[0];
					log.info(eventPath + "------get " + lockName + "--------------");
					locks.get(lockName).latch.countDown();
				}
			} else if (event.getType() == EventType.None && event.getState() == KeeperState.SyncConnected) {
				String addr = InetAddress.getLocalHost().getHostAddress();
				log.info(addr + " register......");
			} else if (event.getType() == EventType.NodeDeleted) {
				log.info(eventPath + "节点删除..........");
				
				if (eventPath.equals(this.leaderPath)) {// 领导节点负责创建任务路径，如果删除需要重新注册

					final ZkAppCenter _zkapp = this;
					new Thread() {
						public void run() {
							try {
								_zkapp.registerApp();
							} catch (Exception e) {
								log.error(e.getMessage(), e);
							}
						}
					}.start();
				}
			} else {

				byte[] btask = this.getZK().getData(eventPath, true, null);
				
				if (btask == null || btask.length == 0) {
					return;
				}

				final String task = new String(btask);// 继续监听
				ThreadPoolUtil.asynRun(new Runnable() {
					
					@Override
					public void run() {
						try {
							if (eventPath.equals(normalTask)) {
								// @markfb
								 doInnerNormalTask(task);
							} else {
								doTask(task);
							}
						} catch (Exception e) {
							log.error(e.getMessage(), e);
						}
					}
				});

				// }
				log.debug("this.appName:" + this.appName);
			}
		} catch (Exception e) {
			log.info(e);
		}
	}

	public void reRegister() {
		sessionTimeout = true;
		synchronized (ZkAppCenter.class) {
			if (sessionTimeout) {
				try {
					//latchControl.wakeupWait();
					if (zk != null) {
						this.zk.close();
						zk = null;
					} 
					
					this.registerApp();
					Thread.sleep(10000);
					// Collection<CountDownLatch> latchs =
					// taskLatchMap.values();
					// registerApp();
					sessionTimeout = false;// 排除故障操作完毕
					log.info("zookeeper failure has been ruled out............");
					//this.latchControl.countDown();// 继续
				} catch (Exception e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
			}
		}

	}

	//************************************************
	// 2014-09-23 改版新加方法
	
	/**
	 * 主要用于新添加的渠道任务生成
	 * 
	 * @param code 渠道code
	 * @throws Exception
	 */
	public void publishZkTaskByChannelCode(String code) throws Exception {
		String taskPath = null;
		Set<String> tasks = getZkTask(code);
		
		for (String task : tasks) {
			taskPath = this.getTaskPath(zk, task);
			zk.getData(taskPath, true, null);
		}
		
		taskSet.addAll(tasks);
	}
	
	private Set<String> getZkTask(String code) {
		Set<String> tasks = new HashSet<String>();
		tasks.add(code + "_CheckInTask");
		tasks.add(code + "_CheckOutTask");
		
		return tasks;
	}
}
